package edu.ilstu.mind.ise.ccp.impl.disruptor;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.MultiThreadedClaimStrategy;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;

import edu.ilstu.mind.ise.ccp.CentralControlProcessor;
import edu.ilstu.mind.ise.ccp.config.ConfigParserUtil;
import edu.ilstu.mind.ise.ccp.logger.Logger;
import edu.ilstu.mind.ise.ccp.message.MessageSender;
import edu.ilstu.mind.ise.ccp.message.hashmap.HashMapMutableMessageFactory;
import edu.ilstu.mind.ise.ccp.mindmodule.MindModule;
import edu.ilstu.mind.ise.ccp.module.Module;

/**
 * Implementation of {@link CentralControlProcessor} that uses the {@link Disruptor}
 * framework to delegate messages to modules
 * 
 * @author Eric Thill
 * 
 */

public class DisruptorCentralControlProcessor implements CentralControlProcessor {

	public static final int DEFAULT_BUFFERSIZE = 1024;

	private final int bufferSize;
	private final WaitStrategy waitStrategy;
	private final Map<String, Module> modules = new HashMap<String, Module>();
	private Logger logger;
	private MindModule mindModule;

	/**
	 * Default Constructor. bufferSize=1024 waitStrategy=BlockingWaitStrategy
	 */
	public DisruptorCentralControlProcessor() {
		this(DEFAULT_BUFFERSIZE, new BlockingWaitStrategy());
	}

	/**
	 * Constructor
	 * 
	 * @param bufferSize
	 *            size of ring buffer. must be a power of 2
	 * @param waitStrategy
	 *            wait strategy to be used by disruptor
	 *            {BLOCKING,YIELDING,SLEEPING,BUSY_SPIN}
	 */
	public DisruptorCentralControlProcessor(int bufferSize, WaitStrategy waitStrategy) {
		this.bufferSize = bufferSize;
		this.waitStrategy = waitStrategy;
		// TODO assert bufferSize is a power of 2
	}

	/**
	 * Constructor that parses configuration from <String,Object> Map
	 * 
	 * @param props
	 *            Properties to be parsed <BR>
	 *            bufferSize=int that is a power of 2 (default=1024) <BR>
	 *            waitStrategy={BLOCKING,YIELDING,SLEEPING,BUSY_SPIN}
	 *            (default=BLOCKING)
	 */
	public DisruptorCentralControlProcessor(Map<String, Object> props) {
		this(ConfigParserUtil.getInt(props, "bufferSize", DEFAULT_BUFFERSIZE), WaitStrategyParser.parseWaitStrategy(ConfigParserUtil
				.getString(props, "waitStrategy", WaitStrategyParser.WAITSTRATEGY_BLOCKING)));
	}

	@Override
	public void setLogger(Logger logger) {
		this.logger = logger;
	}

	@Override
	public void setMindModule(MindModule mindModule) {
		this.mindModule = mindModule;
	}

	@Override
	public void addModule(String name, Module module) {
		// make sure module does not already exist
		if (modules.containsKey(name)) {
			throw new IllegalArgumentException("Cannot add multiple modules with the same name <" + name + ">");
		}
		// add module to module map
		modules.put(name, module);
	}

	@Override
	public void init() {
		logger.info("Initializing " + getClass().getSimpleName());

		// verify setters were called
		if (mindModule == null) {
			throw new IllegalArgumentException("mindModule cannot be null");
		}
		if (modules.size() == 0) {
			throw new IllegalArgumentException("Number of modules must be greater than 0");
		}

		// create disruptor
		int numThreads = modules.size() + 1; // +1 for mind module
		Disruptor<MessageEvent> disruptor = new Disruptor<MessageEvent>(new MessageEventFactory(),
				Executors.newFixedThreadPool(numThreads), new MultiThreadedClaimStrategy(bufferSize), waitStrategy);
		
		// create message sender
		MessageSender messageSender = new DisruptorMessageSender(disruptor.getRingBuffer());

		// create mind module event handler
		MindModuleEventHandler[] mindModuleEventHandlers = new MindModuleEventHandler[] { new MindModuleEventHandler(mindModule) };

		// create module event handlers
		ModuleEventHandler[] moduleEventHandlers = new ModuleEventHandler[modules.size()];
		
		// iterate through all modules to create all of the event handlers
		int i = 0;
		for (Map.Entry<String, Module> moduleEntry : modules.entrySet()) {
			moduleEventHandlers[i] = new ModuleEventHandler(moduleEntry.getKey(), moduleEntry.getValue());
			i++;
		}

		// tell disruptor to handle mind module then all modules
		disruptor.handleEventsWith(mindModuleEventHandlers).then(moduleEventHandlers);

		// start the disruptor
		disruptor.start();
		
		// initialize the mind module
		mindModule.setMessageFactory(new HashMapMutableMessageFactory()); //TODO more efficient factory that reuses objects
		mindModule.setMessageSender(messageSender);
		mindModule.init();
		
		// initialize all modules
		for (Map.Entry<String, Module> moduleEntry : modules.entrySet()) {
			// set message factory and message sender
			//TODO more efficient factory that reuses objects
			moduleEntry.getValue().setMessageFactory(new HashMapMutableMessageFactory(moduleEntry.getKey()));
			moduleEntry.getValue().setMessageSender(messageSender);
			// initialize
			moduleEntry.getValue().init();
		}
		
		logger.info("Initialized");
	}

}

class MessageEventFactory implements EventFactory<MessageEvent> {
	// This is used by the Disruptor to initially populate all of the entries in
	// the ring buffer
	@Override
	public MessageEvent newInstance() {
		return new MessageEvent();
	}
}